package org.huangrui.spark.scala.streaming

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}

import java.util.Properties
import scala.collection.mutable.ListBuffer
import scala.util.Random

/**
 * @author hr
 * @create 2020-12-26 16:37 
 */
object SparkStreaming10_MockData {
  def main(args: Array[String]): Unit = {
    // 生成模拟数据
    // 格式 ：timestamp area city userid adid
    // 含义： 时间戳   区域  城市 用户 广告

    // Application => Kafka => SparkStreaming => Analysis
    val prop: Properties = new Properties()
    // 添加配置
    prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop210:9092,hadoop211:9092,hadoop212:9092")
    prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
    prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
    val producer: KafkaProducer[String, String] = new KafkaProducer[String, String](prop)

    while (true){
      mockdata().foreach{
        data =>{
          // 向Kafka中生成数据
          val record: ProducerRecord[String, String] = new ProducerRecord[String,String]("spark",data)
          producer.send(record)
          println(data)
        }
      }
      Thread.sleep(2000)
    }
  }
  def mockdata() ={
    val list: ListBuffer[String] = ListBuffer[String]()
    val areaList: ListBuffer[String] = ListBuffer[String]("华北", "华东", "华南")
    val cityList: ListBuffer[String] = ListBuffer[String]("北京", "上海", "深圳")
    for (i <- 1 to new Random().nextInt(50)){
      val area: String = areaList(new Random().nextInt(3))
      val city: String = cityList(new Random().nextInt(3))
      val userId: Int = new Random().nextInt(6) + 1
      val adId: Int = new Random().nextInt(6) + 1
      list.append(s"${System.currentTimeMillis()} ${area} ${city} ${userId} ${adId}")
    }
    list
  }
}
